package com.talent.core.protocol.mqtt.handler;

import com.talent.core.protocol.mqtt.TlMqttSession;
import com.talent.core.protocol.mqtt.config.Constant;
import com.talent.core.protocol.mqtt.enums.TlMqttMessageType;
import com.talent.core.protocol.mqtt.enums.TlMqttQos;
import com.talent.core.protocol.mqtt.model.entity.TlMessage;
import com.talent.core.protocol.mqtt.model.fix.TlMqttFixedHead;
import com.talent.core.protocol.mqtt.model.payload.TlMqttConnectPayload;
import com.talent.core.protocol.mqtt.model.request.TlMqttConnectReq;
import com.talent.core.protocol.mqtt.model.request.TlMqttPublishReq;
import com.talent.core.protocol.mqtt.model.response.TlMqttConnack;
import com.talent.core.protocol.mqtt.model.variable.TlMqttConnackVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttConnectVariableHead;
import com.talent.core.protocol.mqtt.store.IMqttStore;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
public class TlMqttConnectHandler extends SimpleChannelInboundHandler<TlMqttConnectReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttConnectHandler.class);

    public IMqttStore mqttStore;

    private static final AtomicInteger MESSAGE_ID = new AtomicInteger(0);

    public TlMqttConnectHandler(IMqttStore mqttStore){
        this.mqttStore = mqttStore;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttConnectReq req) throws Exception {

        Channel channel = ctx.channel();
        logger.debug("进入到【CONNECT】的处理器【{}】",channel);
        TlMqttConnectVariableHead variableHead = req.getVariableHead();
        TlMqttConnectPayload payload = req.getPayload();
        TlMqttFixedHead fixedHead = req.getFixedHead();

        String clientId = payload.getClientId();
        //第一步校验连接是否正确
        //连接标识  如果连接标识不是0 则断开连接
        int reserved = variableHead.getReserved();
        if(reserved!=0){
            logger.info("客户端【{}】的连接标识为【{}】，断开连接",clientId,reserved);
            ctx.channel().close();
        }

        //添加心跳检测
        ctx.pipeline().addLast(new IdleStateHandler(0,0,variableHead.getKeepAlive(), TimeUnit.SECONDS));


        //第二步 设置信息到channel中
        // 主要是客户端标识以及正常断开连接的标识位为false
        channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).set(clientId);
        //断开标志位设备false，表示没有走正常断开的逻辑
        channel.attr(AttributeKey.valueOf(Constant.DISCONNECT)).set(false);


        int cleanSession = variableHead.getCleanSession();
        //第五步 回送消息确定
        TlMqttConnack res = buildMqttAck(cleanSession);
        channel.writeAndFlush(res);

        //第三部 处理session问题
        TlMqttSession session = mqttStore.getSessionByClientId(clientId);
        if(session!=null && session.getCtx()!=null){
            logger.error("客户端【{}】以及存在连接，断开连接",clientId);
            channel.close();
        }
        //如果设置了是1  那么建立连接需要使用新的会话
        if(cleanSession==1 || session==null){
            session =new TlMqttSession();
        }else{
            Set<TlMqttPublishReq> republishs = session.getRepublish();
            for (TlMqttPublishReq republish : republishs) {
                channel.writeAndFlush(republish);
            }

        }
        session.setCtx(ctx);
        session.setCleanSession(cleanSession);
        session.setClientId(clientId);
        session.setQos(TlMqttQos.valueOf(fixedHead.getQos()));
        //将会话存储起来
        mqttStore.putSession(clientId, session);

        //第四个步骤存储遗嘱消息
        int willFlag = variableHead.getWillFlag();
        int willRetain = variableHead.getWillRetain();
        if(willFlag==1){
            int willQos = variableHead.getWillQos();
            String willMessage = payload.getWillMessage();
            String willTopic = payload.getWillTopic();
            //没有消息ID
            int remindLength = Constant.SHORT_LENGTH+willTopic.getBytes().length
                    +Constant.SHORT_LENGTH+ willMessage.getBytes().length-Constant.SHORT_LENGTH;
            TlMqttPublishReq pugMsg = new TlMqttPublishReq(willTopic,TlMqttQos.valueOf(willQos),0,willRetain,
                    willMessage, MESSAGE_ID.incrementAndGet(),remindLength);
            TlMessage message = new TlMessage();
            message.setTopic(willTopic);
            message.setReq(pugMsg);
            message.setQos(willQos);
            message.setTimestamp(System.currentTimeMillis());
            session.setWillMessage(message);
            if(willRetain==1){
                mqttStore.putRetainMessage(willTopic,message);
            }
        }
    }

    private static TlMqttConnack buildMqttAck(int currentSession) {
        TlMqttConnack res = new TlMqttConnack();
        TlMqttFixedHead ackFixedHeader = new TlMqttFixedHead();
        ackFixedHeader.setMessageType(TlMqttMessageType.CONNACK.getValue());
        ackFixedHeader.setLength(Constant.SHORT_LENGTH);
        res.setFixedHead(ackFixedHeader);
        TlMqttConnackVariableHead connackVariableHead = new TlMqttConnackVariableHead();
        connackVariableHead.setCurrentSession(0);
        connackVariableHead.setCode(currentSession);
        res.setVariableHead(connackVariableHead);
        return res;
    }

}
